package com.ry.flink.job7;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;

/**
 * 需求：得到并打印每隔 3 秒钟统计前 3 秒内的相同的 key 的所有的事件
 *
 * 数据有序的情况
 * hadoop,1461756862000
 * hadoop,1461756866000
 * hadoop,1461756872000
 * hadoop,1461756873000
 * hadoop,1461756874000
 * hadoop,1461756876000
 * hadoop,1461756877000
 *
 *
 * window + watermark  观察窗口是如何被触发？
 *
 * 可以解决乱序问题
 *
 * flink,1461756879000
 * flink,1461756871000
 * flink,1461756883000
 *
 *
 *
 *
 */
public class WindowWordCountByWaterMark7 {
    public static void main(String[] args) throws  Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //步骤一：设置时间类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> dataStream = env.socketTextStream("hadoop5", 9999);
        dataStream.map(new MapFunction<String, Tuple2<String,Long>>() {
            @Override
            public Tuple2<String, Long> map(String line) throws Exception {
                String[] fields = line.split(",");
                return new Tuple2<>(fields[0],Long.valueOf(fields[1]));
            }
            //步骤二：获取数据里面的event Time
        }).assignTimestampsAndWatermarks(new EventTimeExtractor())
                .keyBy(0)
                .timeWindow(Time.seconds(3))
                .process(new SumProcessWindowFunction())
                .print().setParallelism(1);
        env.execute("WindowWordCountByWaterMark7");
    }

    public static class SumProcessWindowFunction extends ProcessWindowFunction<Tuple2<String,Long>,String,Tuple,TimeWindow> {
        FastDateFormat format = FastDateFormat.getInstance("HH:mm:ss");

        @Override
        public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Long>> elements, Collector<String> out) {
            System.out.println("处理时间："+format.format(context.currentProcessingTime()));
            //window start time
            System.out.println("window start time："+format.format(context.window().getStart()));

            List<String> reuslt = new ArrayList<>();
            for(Tuple2<String,Long> ele:elements) {
                reuslt.add(ele.toString() + "---"+format.format(ele.f1));
            }
            out.collect(reuslt.toString());
            //window end time
            System.out.println("window start time："+format.format(context.window().getEnd()));
        }
    }

    private static class EventTimeExtractor implements AssignerWithPeriodicWatermarks<Tuple2<String,Long>> {
        FastDateFormat format = FastDateFormat.getInstance("HH:mm:ss");
        //当前窗口内的最大时间
        private long currentMaxEventTime = 0L;
        //最大延迟时间
        private long maxOutOfOrderness = 10000;

        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentMaxEventTime - maxOutOfOrderness);
        }

        @Override
        public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
            /**
             * 获取当前事件的时间：currentEleTime
             * [win1] [win2] [win3]
             * */
            Long currentEleTime = element.f1;
            //计算出在当前窗口内的最大时间
            currentMaxEventTime = Math.max(currentMaxEventTime,currentEleTime);
            System.out.println("event= "+element
            +"@Event Time"+format.format(element.f1)
            +"@Max Event Time" + format.format(currentMaxEventTime)
            +"@Current Watermark" + format.format(getCurrentWatermark().getTimestamp()));
            return currentEleTime;
        }
    }
}
